package com.sskj.run.presenter;

import android.util.Log;

import com.google.gson.Gson;
import com.sskj.common.base.App;
import com.sskj.common.http.WebSocketObserver;
import com.sskj.common.rxbus2.RxBus;
import com.sskj.common.util.DisposUtil;
import com.sskj.lib.Constans;
import com.sskj.run.R;
import com.sskj.run.bean.SocketSendBean;
import com.sskj.run.bean.StockBloomBean;
import com.sskj.run.ui.fragment.RunFragment;

import org.reactivestreams.Publisher;

import java.util.ArrayList;
import java.util.concurrent.TimeUnit;

import io.reactivex.Flowable;
import io.reactivex.Observable;
import io.reactivex.android.schedulers.AndroidSchedulers;
import io.reactivex.disposables.Disposable;
import io.reactivex.functions.Function;
import io.reactivex.schedulers.Schedulers;


public class RunFragmentPresenter extends BasePresenter<RunFragment> {
    private WebSocketObserver stockSocket;
    private Disposable stockDisposable;
    private Disposable timeDisposed;
    int currentMillers = 0;
    private boolean isClose = false;

    public void initWebSocket() {
        if (stockSocket == null && !isClose) {
            stockSocket = httpService.pushBloomType(Constans.DEFAULT_BIBI_COIN);
            stockDisposable = stockSocket
                    .subscribeOn(Schedulers.io())
                    .flatMap((Function<String, Publisher<String>>) text -> {
                        ArrayList<String> strings = new ArrayList<>();
                        strings.add(text);
                        return Flowable.fromIterable(strings);
                    })
                    .filter(s -> !s.startsWith("{\"status\":0"))
                    .map(s -> {
                        System.out.println(s);
                        return new Gson().fromJson(s, StockBloomBean.class);
                    })
                    .observeOn(AndroidSchedulers.mainThread())
                    .subscribe(data -> {
                        mView.send(data);
//                        RxBus.getDefault().post(data);
                        currentMillers = 0;
                        //5秒后没有收到下一次推送，判定为卡住，断开重连
                        if (timeDisposed != null) {
                            DisposUtil.close(timeDisposed);
                        }
                        timeDisposed = Observable.interval(0, 1, TimeUnit.SECONDS)
                                .subscribeOn(Schedulers.newThread())
                                .observeOn(AndroidSchedulers.mainThread())
                                .subscribe(aLong -> {
                                    currentMillers++;
                                    Log.i("out", currentMillers + "");
                                    if (currentMillers == 2) {
//                                stockSocket.send(new Gson().toJson(new SocketSendBean(Constans.DEFAULT_BIBI_COIN)));
                                        Log.i("out", App.INSTANCE.getString(R.string.run_runFragmentPresenter1));
                                        if (stockSocket != null) {
                                            stockSocket.close();
                                            stockSocket = null;
                                        }
                                        DisposUtil.close(stockDisposable);
                                        initWebSocket();
                                    }
                                });
                    });
        }
    }


    public void closeWebSocket() {
        isClose = true;
        DisposUtil.close(timeDisposed);
        if (stockSocket != null) {
            stockSocket.close();
            stockSocket = null;
        }
    }


}
